package cn.com.tt.flink.sql.engine.processor;

import cn.com.tt.flink.runner.AbstractProcessor;
import cn.com.tt.flink.sql.engine.pojo.JobConfig;
import cn.com.tt.flink.sql.engine.pojo.JobUdf;
import cn.com.tt.flink.sql.engine.utils.LambdaUtils;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.functions.UserDefinedFunction;

public class RegisterUdfProcessor extends AbstractProcessor {

    public RegisterUdfProcessor(StreamTableEnvironment tableEnv) {
        super(tableEnv);
    }

    @Override
    public void process() throws Exception {
        JobConfig.getInstance().getUdfList().forEach(LambdaUtils.consumerWrapper(this::registerUdf));
    }

    private void registerUdf(JobUdf udf) throws Exception {
        UserDefinedFunction userDefinedFunction = (UserDefinedFunction) Class.forName(udf.getClassName()).newInstance();

        if (userDefinedFunction instanceof ScalarFunction) {
            tableEnv.createTemporaryFunction(udf.getUdfName(), (ScalarFunction) userDefinedFunction);
        } else if (userDefinedFunction instanceof TableFunction) {
            tableEnv.createTemporaryFunction(udf.getUdfName(), (TableFunction) userDefinedFunction);
        } else if (userDefinedFunction instanceof AggregateFunction) {
            tableEnv.createTemporaryFunction(udf.getUdfName(), (AggregateFunction) userDefinedFunction);
        }

    }

}
